/*
 *
 *  * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 *  * openFuyao is licensed under Mulan PSL v2.
 *  * You can use this software according to the terms and conditions of the Mulan PSL v2.
 *  * You may obtain a copy of Mulan PSL v2 at:
 *  *          http://license.coscl.org.cn/MulanPSL2
 *  * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 *  * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 *  * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *  * See the Mulan PSL v2 for more details.
 *
 */

// package ray config
package ray

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"gorm.io/gorm"

	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"

	"sigs.k8s.io/yaml"

	"openfuyao-rayservice/pkg/constant"
	"openfuyao-rayservice/pkg/utils"
	"openfuyao-rayservice/pkg/utils/httputil"
	"openfuyao-rayservice/pkg/zlog"
)

type RayHeadStatus struct {
	State *corev1.ContainerState `json:"state"`
	Phase string                 `json:"phase"`
}

type CrdInfo struct {
	ID           uint   `gorm:"primarykey;autoIncrement" json:"-"`
	Name         string `gorm:"column:name;type:varchar(255);not null;uniqueIndex:idx_name_ns_crd" json:"name"`
	Namespace    string `gorm:"column:namespace;type:varchar(100);not null;uniqueIndex:idx_name_ns_crd" json:"namespace"`
	CreateBy     string `gorm:"column:create_by;type:varchar(255);not null;comment:creator name" json:"create_by"`
	ErrorInfo    string `gorm:"column:error_info;type:text;comment:error info mssage" json:"error_info"`
	ConfigDetail string `gorm:"column:config_detail;type:text;not null;comment:raycl detail config" json:"config_detail"`
	CreateAt     int64  `gorm:"column:create_at;autoCreateTime;comment:create time" json:"-"`
	UpdateAt     int64  `gorm:"column:update_at;autoUpdateTime;comment:update time" json:"-"`
	RayVersion   string `gorm:"column:ray_version;type:varchar(100);not null;comment:ray vesion" json:"ray_version"`
	Status       int    `gorm:"column:status;type:int(2);comment:status. {0: draft, 1: activate}" json:"-"`
	CrdType      string `gorm:"column:crd_type;type:varchar(100);not null;uniqueIndex:idx_name_ns_crd;comment:kuberay crd type" json:"-"`

	CreateTime        string `gorm:"-:all" json:"create_time"`
	UpdateTime        string `gorm:"-:all" json:"update_time"`
	ClusterStatus     string `gorm:"-:all" json:"cluster_status"`
	RayHeadLogs       string `gorm:"-:all" json:"ray_head_logs"`
	TemplateStatus    string `gorm:"-:all" json:"template_status"`
	ClusterStatusZHCN string `gorm:"-:all" json:"cluster_status_zhcn"`
}

type RayCluster struct {
	*CrdInfo
	GcsPort       string        `gorm:"-:all" json:"gcs_port"`
	ClusterIP     string        `gorm:"-:all" json:"cluster_ip"`
	DashboardPort string        `gorm:"-:all" json:"dashboard_port"`
	RayHeadStatus RayHeadStatus `gorm:"-:all" json:"ray_head_status"`
}

type RayService struct {
	*CrdInfo
	DashboardAddress string `gorm:"-:all" json:"dashboard_address"`
}

type RayJob struct {
	*CrdInfo
	JobId               string `gorm:"-:all" json:"job_id"`
	RayClusterName      string `gorm:"-:all" json:"raycluster_name"`
	DashboardAddress    string `gorm:"-:all" json:"dashboard_address"`
	JobDeploymentStatus string `gorm:"-:all" json:"job_deployment_status"`
}

func (CrdInfo) TableName() string {
	return "kuberay_crd"
}

func (v rayClient) MethodInitOperation() {
	table := CrdInfo{}
	zlog.Infof("init table %s ...", table.TableName())

	err := v.dbConnection.Table(table.TableName()).Migrator().AutoMigrate(&table)
	if err != nil {
		zlog.FormatFatal("Create Table: `kuberay_crd` error, %v: \n", err)
	}

}

func (v rayClient) MethodAllOverview() (*httputil.ResponseJson, int) {

	rayclusters, err := v.rayClientset.RayV1().RayClusters("").List(context.TODO(), metav1.ListOptions{
		LabelSelector: fmt.Sprintf("%s=%s", constant.RayClusterCreateBy, constant.ConsoleServiceDefaultOrgName),
	})
	if err != nil {
		zlog.Errorf("Get system RayClusters[Fuyao] failed, %v", err)
		return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
	}

	rayjobs, err := v.rayClientset.RayV1().RayJobs("").List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		zlog.Errorf("Get system rayjobs failed, %v", err)
		return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
	}

	rayservices, err := v.rayClientset.RayV1().RayServices("").List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		zlog.Errorf("Get system rayservices failed, %v", err)
		return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
	}

	total, err := v.rayClientset.RayV1().RayClusters("").List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		zlog.Errorf("Get system RayClusters[Total] failed, %v", err)
		return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
	}

	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{
			"rayclusters": len(rayclusters.Items),
			"rayservices": len(rayservices.Items),
			"rayjobs":     len(rayjobs.Items),
			"total":       len(total.Items),
		},
	}, http.StatusOK

}

func (v rayClient) MethodCreateCrd(crd, data, createBy string, isStart bool) (*httputil.ResponseJson, int) {
	return v.doCreateCrd(crd, data, createBy, isStart, false)
}

func (v rayClient) doCreateCrd(crd, data, createBy string, isStart, fromRestart bool) (*httputil.ResponseJson, int) {
	rc := CrdInfo{
		CreateBy:     createBy,
		ErrorInfo:    "",
		ConfigDetail: data,
		CrdType:      crd,
		Status:       0,
	}
	var crdResource interface{} = rayv1.RayCluster{}
	switch crd {
	case constant.RayCluster:
		rayClusterInstance := rayv1.RayCluster{}
		err := yaml.Unmarshal([]byte(data), &rayClusterInstance)
		if err != nil {
			zlog.Errorf("Get user rayClusterInstance failed, %v", err)
			return httputil.GetClientFailureResponseJsonWithData(err.Error()), http.StatusBadRequest
		}
		rc.Name = rayClusterInstance.Name
		rc.Namespace = getDefaultNs(rayClusterInstance.Namespace, constant.Default)
		rayClusterInstance.Namespace = rc.Namespace
		rc.RayVersion = rayClusterInstance.Spec.RayVersion
		crdResource = &rayClusterInstance
	case constant.RayService:
		rayServiceInstance := rayv1.RayService{}
		err := yaml.Unmarshal([]byte(data), &rayServiceInstance)
		if err != nil {
			zlog.Errorf("Get user rayServiceInstance failed, %v", err)
			return httputil.GetClientFailureResponseJsonWithData(err.Error()), http.StatusBadRequest
		}
		rc.Name = rayServiceInstance.Name
		rc.Namespace = getDefaultNs(rayServiceInstance.Namespace, constant.Default)
		rayServiceInstance.Namespace = rc.Namespace
		rc.RayVersion = rayServiceInstance.Spec.RayClusterSpec.RayVersion
		crdResource = &rayServiceInstance
	case constant.RayJob:
		rayJobInstance := rayv1.RayJob{}
		err := yaml.Unmarshal([]byte(data), &rayJobInstance)
		if err != nil {
			zlog.Errorf("Get user rayJobInstance failed, %v", err)
			return httputil.GetClientFailureResponseJsonWithData(err.Error()), http.StatusBadRequest
		}
		rc.Name = rayJobInstance.Name
		rc.Namespace = getDefaultNs(rayJobInstance.Namespace, constant.Default)
		rayJobInstance.Namespace = rc.Namespace
		rc.RayVersion = rayJobInstance.Spec.RayClusterSpec.RayVersion
		crdResource = &rayJobInstance
	default:
		zlog.Errorf("Unknown Resource Found ..")
		return httputil.GetDefaultClientFailureResponseJson(), http.StatusBadRequest
	}

	if !fromRestart {
		crdInfo := CrdInfo{}
		v.dbConnection.Where("name = ?", rc.Name).
			Where("crd_type = ?", crd).
			Where("namespace = ?", rc.Namespace).First(&crdInfo)
		if crdInfo.Name != "" {
			return &httputil.ResponseJson{
				Code: constant.ClientError,
				Msg:  "Resource already exists!",
				Data: []any{},
			}, http.StatusOK
		}
	}

	v.dbConnection.Transaction(func(tx *gorm.DB) error {
		if v.dbConnection.Model(&rc).Where("name = ?", rc.Name).
			Where("crd_type = ?", crd).
			Where("namespace = ?", rc.Namespace).Updates(&rc).RowsAffected == 0 {
			createInfo := v.dbConnection.Create(&rc)
			if createInfo.Error != nil {
				zlog.Errorf("Create rayClusterInstance failed, %v", createInfo.Error)
				return createInfo.Error
			}
		} else {
			v.dbConnection.Model(&rc).Where("name = ?", rc.Name).
				Where("crd_type = ?", crd).
				Where("namespace = ?", rc.Namespace).First(&rc)
		}
		return nil
	})
	if isStart {
		var err error
		switch crd {
		case constant.RayCluster:
			rayClusterInstance := crdResource.(*rayv1.RayCluster)
			if rayClusterInstance.Labels == nil {
				rayClusterInstance.Labels = make(map[string]string)
			}
			rayClusterInstance.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName
			rayClusterInstance.Labels[constant.RayClusterCreateBy] = constant.ConsoleServiceDefaultOrgName

			headPod := rayClusterInstance.Spec.HeadGroupSpec.Template
			if headPod.Labels == nil {
				headPod.Labels = make(map[string]string)
			}
			headPod.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName

			_, err = v.rayClientset.RayV1().RayClusters(rayClusterInstance.Namespace).
				Create(context.TODO(), rayClusterInstance, metav1.CreateOptions{})
		case constant.RayService:
			rayServiceInstance := crdResource.(*rayv1.RayService)
			if rayServiceInstance.Labels == nil {
				rayServiceInstance.Labels = make(map[string]string)
			}
			rayServiceInstance.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName

			headPod := rayServiceInstance.Spec.RayClusterSpec.HeadGroupSpec.Template
			if headPod.Labels == nil {
				headPod.Labels = make(map[string]string)
			}
			headPod.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName

			_, err = v.rayClientset.RayV1().RayServices(rayServiceInstance.Namespace).
				Create(context.TODO(), rayServiceInstance, metav1.CreateOptions{})
		case constant.RayJob:
			rayJobInstance := crdResource.(*rayv1.RayJob)
			if rayJobInstance.Labels == nil {
				rayJobInstance.Labels = make(map[string]string)
			}
			rayJobInstance.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName

			headPod := rayJobInstance.Spec.RayClusterSpec.HeadGroupSpec.Template
			if headPod.Labels == nil {
				headPod.Labels = make(map[string]string)
			}
			headPod.Labels[constant.KuberayCrdCreateBy] = constant.ConsoleServiceDefaultOrgName

			_, err = v.rayClientset.RayV1().RayJobs(rayJobInstance.Namespace).
				Create(context.TODO(), rayJobInstance, metav1.CreateOptions{})
		default:
			zlog.Errorf("Unknown Resource Found ..")
			return httputil.GetDefaultClientFailureResponseJson(), http.StatusBadRequest
		}

		if err != nil {
			zlog.Errorf("Create %s failed, %v", crd, err)
			rc.ErrorInfo = err.Error()
			v.dbConnection.Save(&rc)
			return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
		}
		rc.Status = 1
		v.dbConnection.Save(&rc)
	}
	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{},
	}, http.StatusOK
}

func (v rayClient) MethodUpdateCrd(crd, ns, data, name string) (*httputil.ResponseJson, int) {
	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{},
	}, http.StatusOK
}

func (v rayClient) MethodDeleteCrd(crd, ns, name string) (*httputil.ResponseJson, int) {
	var resourceName string
	var isRunning bool = false
	switch crd {
	case constant.RayCluster:
		raycluster, _ := v.rayClientset.RayV1().RayClusters(ns).Get(context.TODO(), name, metav1.GetOptions{})
		resourceName = raycluster.Name
		isRunning = raycluster.Status.State == rayv1.Ready
	case constant.RayService:
		rayservice, _ := v.rayClientset.RayV1().RayServices(ns).Get(context.TODO(), name, metav1.GetOptions{})
		resourceName = rayservice.Name
		isRunning = rayservice.Status.ServiceStatus == rayv1.Running
	case constant.RayJob:
		rayjob, _ := v.rayClientset.RayV1().RayJobs(ns).Get(context.TODO(), name, metav1.GetOptions{})
		resourceName = rayjob.Name
		isRunning = rayjob.Status.JobDeploymentStatus == rayv1.JobDeploymentStatusRunning
	default:
		zlog.Errorf("Unknown Resource Found ..")
		return httputil.GetDefaultClientFailureResponseJson(), http.StatusBadRequest
	}
	if resourceName != "" {
		if isRunning {
			return &httputil.ResponseJson{
				Code: constant.ClientError,
				Msg:  "不允许删除正在运行的资源!",
				Data: map[string]any{},
			}, http.StatusOK
		}
		// other delete k8s resource
		v.MethodStopCrd(crd, ns, name)
	}
	delRes := v.dbConnection.Model(&CrdInfo{}).Where("name = ?", name).
		Where("crd_type = ?", crd).
		Where("namespace = ?", ns).Delete(&CrdInfo{})
	if delRes.RowsAffected == 0 {
		zlog.Errorf("Delete crd failed, crd: %s, ns: %s, name: %s", crd, ns, name)
		return httputil.GetResponseJson(
			constant.ClientError,
			"资源不存在!",
			map[string]any{}), http.StatusOK
	}
	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{},
	}, http.StatusOK
}

func (v rayClient) MethodStopCrd(crd, ns, name string) (*httputil.ResponseJson, int) {

	var err error
	switch crd {
	case constant.RayCluster:
		err = v.rayClientset.RayV1().RayClusters(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
	case constant.RayService:
		err = v.rayClientset.RayV1().RayServices(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
	case constant.RayJob:
		err = v.rayClientset.RayV1().RayJobs(ns).Delete(context.TODO(), name, metav1.DeleteOptions{})
	default:
		zlog.Errorf("Unknown Resource Found ..")
		return httputil.GetDefaultClientFailureResponseJson(), http.StatusBadRequest
	}
	if err != nil {
		zlog.Errorf("Stop Crd failed, type: %s, name: %s,  error: %v", crd, name, err)
		return httputil.GetServerFailureResponseJsonWithData(err.Error()), http.StatusInternalServerError
	}
	v.dbConnection.Model(&CrdInfo{}).
		Where("name = ?", name).
		Where("crd_type = ?", crd).
		Where("namespace = ?", ns).
		Update("error_info", "")
	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{},
	}, http.StatusOK
}

func (v rayClient) MethodRestartCrd(crd, ns, name string) (*httputil.ResponseJson, int) {
	crdInfo := CrdInfo{}
	v.dbConnection.Where("name = ?", name).
		Where("namespace = ?", ns).
		Where("crd_type = ?", crd).
		First(&crdInfo)
	if crdInfo.Name == "" {
		zlog.Errorf("Restart crd failed, name: %s", name)
		return httputil.GetDefaultServerFailureResponseJson(), http.StatusInternalServerError
	}
	_, err := v.doGetCrd(crd, ns, name)
	if err == nil { // Already exists resource
		return &httputil.ResponseJson{
			Code: constant.ClientError,
			Msg:  "资源正在运行中!",
			Data: []any{},
		}, http.StatusOK
	}
	return v.doCreateCrd(crd, crdInfo.ConfigDetail, crdInfo.CreateBy, true, true)
}

func (v rayClient) MethodGetCrd(crd, ns, name string) (*httputil.ResponseJson, int) {
	crdInfo := CrdInfo{}
	v.dbConnection.Where("name = ?", name).
		Where("crd_type = ?", crd).
		Where("namespace = ?", ns).First(&crdInfo)
	if crdInfo.Name == "" {
		return &httputil.ResponseJson{
			Code: constant.ResourceNotFound,
			Msg:  "Not Found",
			Data: []any{},
		}, http.StatusOK
	}
	var data interface{} = &crdInfo

	crdResource, err := v.doGetCrd(crd, ns, name)

	if err != nil {
		zlog.Errorf("Get crd failed, name: %s,  error: %v", name, err)
		if crdInfo.ErrorInfo != "" {
			crdInfo.ClusterStatus = constant.Failed
		} else {
			crdInfo.ClusterStatus = constant.Suspended
		}
		crdInfo.TemplateStatus = constant.Template
	} else {
		crdInfo.TemplateStatus = constant.Instance
		var err error
		var rayV1RayCluster *rayv1.RayCluster = &rayv1.RayCluster{}
		switch crd {
		case constant.RayCluster:
			rayCluster := RayCluster{
				CrdInfo: &crdInfo,
			}
			transferRayClusterToEntity(crdResource.(*rayv1.RayCluster), &rayCluster)
			rayV1RayCluster = crdResource.(*rayv1.RayCluster)
			stateMap := v.getPodStatus(rayCluster.Namespace, &[]string{rayCluster.Name})
			rayCluster.RayHeadStatus = *(*stateMap)[rayCluster.Name]
			rayCluster.ClusterStatus = string(rayV1RayCluster.Status.State)
			data = &rayCluster

		case constant.RayService:
			rayService := RayService{
				CrdInfo: &crdInfo,
			}
			rayV1RayService := crdResource.(*rayv1.RayService)
			rayV1RayCluster, _ = v.rayClientset.RayV1().RayClusters(ns).
				Get(context.TODO(), rayV1RayService.Status.ActiveServiceStatus.RayClusterName, metav1.GetOptions{})
			rayService.DashboardAddress = fmt.Sprintf("%s:%s",
				rayV1RayService.Status.ActiveServiceStatus.RayClusterStatus.Head.ServiceIP,
				rayV1RayService.Status.ActiveServiceStatus.RayClusterStatus.Endpoints["dashboard"])
			rayService.ClusterStatus = string(rayV1RayService.Status.ServiceStatus)
			data = &rayService

		case constant.RayJob:
			rayJob := RayJob{
				CrdInfo: &crdInfo,
			}
			rayV1RayJob := crdResource.(*rayv1.RayJob)
			rayV1RayCluster, _ = v.rayClientset.RayV1().RayClusters(ns).
				Get(context.TODO(), rayV1RayJob.Status.RayClusterName, metav1.GetOptions{})
			rayJob.JobId = rayV1RayJob.Status.JobId
			rayJob.RayClusterName = rayV1RayJob.Status.RayClusterName
			rayJob.DashboardAddress = fmt.Sprintf("%s:%s",
				rayV1RayCluster.Status.Head.PodIP,
				rayV1RayCluster.Status.Endpoints["dashboard"])
			rayJob.JobDeploymentStatus = string(rayV1RayJob.Status.JobDeploymentStatus)
			rayJob.ClusterStatus = string(rayV1RayJob.Status.JobDeploymentStatus)
			data = &rayJob
		default:
			zlog.Errorf("Unknown Resource Found ..")
		}

		podList, err := v.clientset.CoreV1().Pods(ns).List(context.TODO(), metav1.ListOptions{
			LabelSelector: fmt.Sprintf("ray.io/cluster=%s", rayV1RayCluster.Name),
		})
		if err != nil {
			zlog.Errorf("Get rayCluster Pod failed, name: %s,  error: %v", name, err)
		}
		if len(podList.Items) != 0 {
			crdInfo.RayHeadLogs = v.fetchPodLog(rayV1RayCluster.Namespace, rayV1RayCluster.Name, podList.Items[0].Name,
				rayV1RayCluster.Spec.HeadGroupSpec.Template.Spec.Containers[0].Name)
		}
	}
	normalizeEntity(&crdInfo)
	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: data,
	}, http.StatusOK
}

func (v rayClient) doGetCrd(crd, ns, name string) (interface{}, error) {
	var crdResource interface{} = rayv1.RayCluster{}
	var err error
	switch crd {
	case constant.RayCluster:
		crdResource, err = v.rayClientset.RayV1().RayClusters(ns).Get(context.TODO(), name, metav1.GetOptions{})
	case constant.RayService:
		crdResource, err = v.rayClientset.RayV1().RayServices(ns).Get(context.TODO(), name, metav1.GetOptions{})
	case constant.RayJob:
		crdResource, err = v.rayClientset.RayV1().RayJobs(ns).Get(context.TODO(), name, metav1.GetOptions{})
	default:
		zlog.Errorf("Unknown Resource Found ..")
		return nil, errors.New("unknown resource found")
	}
	return crdResource, err
}

func (v rayClient) MethodListMetaCrd(crd string) (*httputil.ResponseJson, int) {
	data := make(map[string]any)
	var createBy []string
	v.dbConnection.Model(&CrdInfo{}).Distinct().Pluck("create_by", &createBy)
	data["createBy"] = createBy

	switch crd {
	case constant.RayCluster:
		data["status"] = constant.ClusterStatusMap
		var rayVersion []string
		v.dbConnection.Model(&CrdInfo{}).Distinct().Pluck("ray_version", &rayVersion)
		data["rayVersion"] = rayVersion
	case constant.RayService:
		data["status"] = constant.ServiceStatusMap
	case constant.RayJob:
		data["status"] = constant.JobDeploymentStatusMap
	default:
		zlog.Errorf("Unknown Resource Found ..")
		return httputil.GetDefaultClientFailureResponseJson(), http.StatusBadRequest
	}

	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: data,
	}, http.StatusOK
}

func (v rayClient) MethodListCrd(crd string, req *CrdListRequest) (*httputil.ResponseJson, int) {
	var crdInfos []CrdInfo

	db := v.dbConnection.Order("update_at desc")
	db = db.Where("crd_type = ?", crd)
	if req.Name != "" {
		db = db.Where("name like ?", "%"+req.Name+"%")
	}
	if req.RayVersion != "" {
		db = db.Where("ray_version = ?", req.RayVersion)
	}
	if req.CreateBy != "" {
		db = db.Where("create_by = ?", req.CreateBy)
	}
	db.Find(&crdInfos)
	lens := len(crdInfos)
	var data []interface{}

	if lens > 0 {
		keyMappings := make(map[string]interface{})
		// prehandle
		utils.MapInPlace(crdInfos, func(item *CrdInfo) {
			switch crd {
			case constant.RayCluster:
				keyMappings[item.Name+":"+item.Namespace] = &RayCluster{
					CrdInfo: item,
				}
			case constant.RayService:
				keyMappings[item.Name+":"+item.Namespace] = &RayService{
					CrdInfo: item,
				}
			case constant.RayJob:
				keyMappings[item.Name+":"+item.Namespace] = &RayJob{
					CrdInfo: item,
				}
			default:
				zlog.Errorf("Unknown Resource Found ..")
			}
			if item.ErrorInfo != "" {
				item.ClusterStatus = constant.Failed
			} else {
				item.ClusterStatus = constant.Suspended
			}
			item.TemplateStatus = constant.Template
			item.ConfigDetail = ""
		})

		switch crd {
		case constant.RayCluster:
			rayClusters, err := v.rayClientset.RayV1().RayClusters("").List(context.TODO(), metav1.ListOptions{})
			if err != nil {
				zlog.Errorf("List rayCluster failed,  error: %v", err)
				return httputil.GetDefaultServerFailureResponseJson(), http.StatusInternalServerError
			}

			aliveNamespacedRaycls := make(map[string][]string)
			utils.MapInPlace(rayClusters.Items, func(item *rayv1.RayCluster) {
				if _, ok := aliveNamespacedRaycls[item.Namespace]; !ok {
					aliveNamespacedRaycls[item.Namespace] = []string{}
				}
				aliveNamespacedRaycls[item.Namespace] = append(aliveNamespacedRaycls[item.Namespace], item.Name)
				key := item.Name + ":" + item.Namespace
				if v, ok := keyMappings[key]; ok {
					rc := v.(*RayCluster)
					rc.TemplateStatus = constant.Instance
					rc.ClusterStatus = string(item.Status.State)
					rc.DashboardPort = item.Status.Endpoints["dashboard"]
					rc.GcsPort = item.Status.Endpoints["gcs"]
					rc.ClusterIP = item.Status.Head.PodIP
				}
			})

			// Aggregation RayCluster HeadPod Status Infomation By Namespace
			for ns, raycls := range aliveNamespacedRaycls {
				stateMap := v.getPodStatus(ns, &raycls)
				utils.MapInPlace(raycls, func(rayclusterName *string) {
					key := *rayclusterName + ":" + ns
					if v, ok := keyMappings[key]; ok {
						rc := v.(*RayCluster)
						rc.RayHeadStatus = *(*stateMap)[*rayclusterName]
					}
				})
			}

		case constant.RayService:
			rayServices, err := v.rayClientset.RayV1().RayServices("").List(context.TODO(), metav1.ListOptions{})
			if err != nil {
				zlog.Errorf("List rayServices failed,  error: %v", err)
				return httputil.GetDefaultServerFailureResponseJson(), http.StatusInternalServerError
			}

			utils.MapInPlace(rayServices.Items, func(item *rayv1.RayService) {
				key := item.Name + ":" + item.Namespace
				if v, ok := keyMappings[key]; ok {
					rs := v.(*RayService)
					rs.TemplateStatus = constant.Instance
					rs.ClusterStatus = string(item.Status.ServiceStatus)
					rs.DashboardAddress = fmt.Sprintf("%s:%s",
						item.Status.ActiveServiceStatus.RayClusterStatus.Head.ServiceIP,
						item.Status.ActiveServiceStatus.RayClusterStatus.Endpoints["dashboard"])
					// For filter status
					if _, ok := constant.ServiceFailedMap[rayv1.ServiceStatus(rs.ClusterStatus)]; ok {
						rs.ClusterStatus = constant.Failed
					}
				}
			})

		case constant.RayJob:
			rayJobs, err := v.rayClientset.RayV1().RayJobs("").List(context.TODO(), metav1.ListOptions{})
			if err != nil {
				zlog.Errorf("List rayServices failed,  error: %v", err)
				return httputil.GetDefaultServerFailureResponseJson(), http.StatusInternalServerError
			}

			utils.MapInPlace(rayJobs.Items, func(item *rayv1.RayJob) {
				key := item.Name + ":" + item.Namespace
				if elem, ok := keyMappings[key]; ok {
					rj := elem.(*RayJob)
					rayV1RayCluster, _ := v.rayClientset.RayV1().RayClusters(rj.Namespace).
						Get(context.TODO(), item.Status.RayClusterName, metav1.GetOptions{})
					rj.TemplateStatus = constant.Instance
					rj.ClusterStatus = string(item.Status.JobDeploymentStatus)
					rj.JobId = item.Status.JobId
					rj.RayClusterName = item.Status.RayClusterName
					rj.DashboardAddress = fmt.Sprintf("%s:%s",
						rayV1RayCluster.Status.Head.PodIP,
						rayV1RayCluster.Status.Endpoints["dashboard"])
					rj.JobDeploymentStatus = string(item.Status.JobDeploymentStatus)
				}
			})
		default:
			zlog.Errorf("Unknown Resource Found ..")
		}

		utils.MapInPlace(crdInfos, func(r *CrdInfo) {
			if req.Status == "" || strings.EqualFold(req.Status, r.ClusterStatus) {
				normalizeEntity(r)
				data = append(data, keyMappings[r.Name+":"+r.Namespace])
			}
		})
	}

	return &httputil.ResponseJson{
		Code: constant.Success,
		Msg:  "success",
		Data: map[string]any{
			"items": data,
			"total": len(data),
		},
	}, http.StatusOK
}

func (v rayClient) fetchPodLog(ns, crdName, podName, containerName string) string {
	// 设置日志选项
	logOptions := &corev1.PodLogOptions{
		Container: containerName,    // 容器名称
		Follow:    false,            // 是否持续跟踪日志（类似 `kubectl logs -f`）
		TailLines: &[]int64{100}[0], // 获取最后 100 行日志
	}

	// 请求日志流
	req := v.clientset.CoreV1().Pods(ns).GetLogs(podName, logOptions) // podList.Items[0].Name
	podLogs, err := req.Stream(context.TODO())
	if err != nil {
		zlog.Errorf("Get rayCluster head-pod logs failed, name: %s,  error: %v", crdName, err)
		return ""
	}
	defer podLogs.Close()

	// 读取日志内容
	logBytes, err := io.ReadAll(podLogs)
	if err != nil {
		zlog.Errorf("Read rayCluster head-pod logs failed, name: %s,  error: %v", crdName, err)
		return ""
	}
	return string(logBytes)
}

func (v rayClient) getPodStatus(ns string, raycls *[]string) *map[string]*RayHeadStatus {

	statusMap := make(map[string]*RayHeadStatus)

	for _, v := range *raycls {
		statusMap[v] = &RayHeadStatus{
			State: nil,
			Phase: "Not Ready",
		}
	}

	pods, err := v.clientset.CoreV1().Pods(ns).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			"ray.io/node-type": "head",
		}).String(),
	})

	if err != nil {
		zlog.Errorf("List head-pod error %v", err)
		return &statusMap
	}

	for _, item := range pods.Items {
		var podClusterName string
		var ok bool
		if podClusterName, ok = item.Labels["ray.io/cluster"]; !ok {
			continue
		}
		if rayHeadStatus, ok := statusMap[podClusterName]; ok {
			rayHeadStatus.Phase = string(item.Status.Phase)
			cs := len(item.Status.ContainerStatuses)
			if cs > 0 {
				rayHeadStatus.State = &item.Status.ContainerStatuses[cs-1].State
			}
		}
	}

	return &statusMap
}

func transferRayClusterToEntity(r1 *rayv1.RayCluster, r2 *RayCluster) {
	r2.ID = 0
	r2.DashboardPort = r1.Status.Endpoints["dashboard"]
	r2.GcsPort = r1.Status.Endpoints["gcs"]
	r2.ClusterIP = r1.Status.Head.PodIP
}

func normalizeEntity(r *CrdInfo) {
	if r.CreateAt != 0 {
		r.CreateTime = utils.UnixFmtTime(r.CreateAt)
	}
	if r.UpdateAt != 0 {
		r.UpdateTime = utils.UnixFmtTime(r.UpdateAt)
	}
	switch r.CrdType {
	case constant.RayCluster:
		if v, ok := constant.ClusterStatusMap[rayv1.ClusterState(r.ClusterStatus)]; ok {
			r.ClusterStatusZHCN = v
		}
	case constant.RayService:
		if v, ok := constant.ServiceStatusMap[rayv1.ServiceStatus(r.ClusterStatus)]; ok {
			r.ClusterStatusZHCN = v
		}
		if _, ok := constant.ServiceFailedMap[rayv1.ServiceStatus(r.ClusterStatus)]; ok {
			r.ClusterStatus = constant.Failed
		}
	case constant.RayJob:
		if v, ok := constant.JobDeploymentStatusMap[rayv1.JobDeploymentStatus(r.ClusterStatus)]; ok {
			r.ClusterStatusZHCN = v
		}
	default:
		zlog.Errorf("Unknown Resource Found ..")
	}

	if v, ok := constant.GlobalStatusMap[r.ClusterStatus]; ok {
		r.ClusterStatusZHCN = v
	}

	if r.ClusterStatusZHCN == "" {
		r.ClusterStatusZHCN = "未知"
	}
}

func getDefaultNs(ns, defaultNs string) string {
	if ns == "" {
		return defaultNs
	}
	return ns
}
